package com.nl.pulsardemo.transaction;

import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.api.transaction.Transaction;

import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author : ZhaoY
 * @description :
 * @date : 2021-02-3 15:55
 **/
@Slf4j
public class ProducerDemo {
    //Pulsar集群中broker的serviceUrl
//    private static String brokerServiceUrl = "pulsar://10.1.8.68:6650,10.1.8.64:6650,10.1.8.69:6650";
    private static String brokerServiceUrl = "pulsar://10.1.8.68:6650";
    //    //指定topic name
    private static String topicName = "persistent://public/default/my-topics";

    public static void main(String[] args) throws PulsarClientException {
        Transaction txn = null;
        //构造Pulsar client
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(brokerServiceUrl)
//                .enableTransaction(true)
                .build();
        try {

            //创建producer
            Producer<String> producer = client.newProducer(Schema.STRING)
                    .topic(topicName)
                    .enableBatching(true)//是否开启批量处理消息，默认true,需要注意的是enableBatching只在异步发送sendAsync生效，同步发送send失效。因此建议生产环境若想使用批处理，则需使用异步发送，或者多线程同步发送
                    .compressionType(CompressionType.LZ4)//消息压缩（四种压缩方式：LZ4，ZLIB，ZSTD，SNAPPY），consumer端不用做改动就能消费，开启后大约可以降低3/4带宽消耗和存储（官方测试）
                    .messageRouter(new com.nl.pulsardemo.pool.config.commons.MyRouteMode())//自定义消息路由
                    .producerName(UUID.randomUUID().toString())//定义生产者名
//                    .messageRouter(new MyRouteMode())//自定义消息路由
                    .create();
            //构建事务
            txn = client
                    .newTransaction()
                    .withTransactionTimeout(5, TimeUnit.MINUTES)
                    .build()
                    .get();


            //创建consumer
            Consumer<String> consumer = client.newConsumer(Schema.STRING)
                    .topic(topicName)
                    .subscriptionName("my-transaction")
                    .subscriptionType(SubscriptionType.Exclusive)//指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式
                    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)//指定从哪里开始消费还有Latest，valueof可选，默认Latest
                    .negativeAckRedeliveryDelay(60, TimeUnit.SECONDS)//指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                    .subscribe();

            //事务消息发送
            producer.newMessage(txn).value("Transaction Test").sendAsync();

            Message<String> message = consumer.receive();
            System.err.println("-=-=-=:" + message.getValue());
            consumer.acknowledgeAsync(message.getMessageId(), txn);

            //Commit transactions
            txn.commit().get();
            //Abort transaction
            txn.abort().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }


}
